package com.xxl.job.admin.core.event;

import com.alibaba.fastjson.JSON;
import com.google.common.eventbus.AllowConcurrentEvents;
import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.Subscribe;
import com.xuxueli.api.exception.JobErrorCode;
import com.xuxueli.api.exception.JobException;
import com.xxl.job.admin.core.conf.BasicThreadFactory;
import com.xxl.job.admin.core.conf.EventBusFactory;
import com.xxl.job.admin.core.model.XxlJobInfo;
import com.xxl.job.admin.service.XxlJobService;
import com.xxl.job.core.biz.model.ReturnT;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * description:
 *
 * <p>
 * Date: 2018/12/27
 * <p>
 * Time: 14:32
 *
 * @author zhangxin
 */
@Component
public class XxlJobInfoAddStartSubscriber {
    @Resource(name = EventBusFactory.JOB_INFO_BATCH_ADD_START_BEAN_NAME)
    private AsyncEventBus batchAddStartEventBus;
    @Autowired
    private XxlJobService xxlJobService;
    private ExecutorService executor;

    @PostConstruct
    public void init() {
        batchAddStartEventBus.register(this);
        initExecutor();
    }

    private void initExecutor(){
        executor = Executors.newFixedThreadPool(5);
    }

    @Subscribe
    @AllowConcurrentEvents
    public void handle(XxlJobInfoAddStartEvent event){
        if(event.getJobInfo() == null || event.getParams() == null || event.getParams().isEmpty()){
            return;
        }

        XxlJobInfo jobInfo = event.getJobInfo();

        for(final String params: event.getParams()){
            final XxlJobInfo job = JSON.parseObject(JSON.toJSONString(jobInfo), XxlJobInfo.class);

            executor.execute(new Runnable(){
                @Override
                public void run() {
                    job.setExecutorParam(params);

                    ReturnT<String> addResult = xxlJobService.add(job);
                    if(addResult.getCode() != ReturnT.SUCCESS_CODE){
                        throw new JobException(JobErrorCode.SERVER_ERROR, addResult.getMsg());
                    }

                    ReturnT<String> startResult = xxlJobService.start(Integer.valueOf(addResult.getContent()).intValue());
                    if(startResult.getCode() != ReturnT.SUCCESS_CODE){
                        throw new JobException(JobErrorCode.SERVER_ERROR, startResult.getMsg());
                    }
                }
            });
        }
    }
}
